package com.example.mq.mqserver.core;

import com.example.mq.common.Consumer;
import com.example.mq.common.ConsumerEnv;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.VirtualHost;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

public class ConsumerManager {
    private VirtualHost parent;
    private ExecutorService workerPool= Executors.newFixedThreadPool(4);
    private BlockingQueue<String> tokenQueue=new LinkedBlockingQueue<>();
    private Thread scannerThread=null;

    public ConsumerManager(VirtualHost p){
        parent=p;
        scannerThread=new Thread(()->{
           while (true){
               try {
                   String queueName = tokenQueue.take();
                   MSGQueue queue=parent.getMemoryDataCenter().getQueue(queueName);
                   if(queue==null){
                       throw new MqException("[ConsumerManager] 取令牌后发现，该队列名不存在！queueName="+queueName);
                   }
                   synchronized (queue){
                       consumeMessage(queue);
                   }
               }catch (InterruptedException | MqException e){
                   e.printStackTrace();
               }
           }
        });
        scannerThread.setDaemon(true);
        scannerThread.start();
    }

    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        MSGQueue queue=parent.getMemoryDataCenter().getQueue(queueName);
        if(queue==null){
            throw new MqException("[ConsumerManager] 队列不存在！queueName="+queueName);
        }
        ConsumerEnv consumerEnv=new ConsumerEnv(consumerTag,queueName,autoAck,consumer);
        synchronized (queue){
            queue.addConsumerEnv(consumerEnv);
            int n=parent.getMemoryDataCenter().getMessageCount(queueName);
            for (int i = 0; i < n; i++) {
                consumeMessage(queue);
            }
        }
    }

    private void consumeMessage(MSGQueue queue) {
        ConsumerEnv luckyDog=queue.chooseConsumer();
        if(luckyDog==null){
            return;
        }
        Message message=parent.getMemoryDataCenter().pollMessage(queue.getName());
        if(message==null){
            return;
        }
        workerPool.submit(()->{
           try {
               parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(), message);
               luckyDog.getConsumer().handleDelivery(luckyDog.getConsumerTag(),message.getBasicProperties(),message.getBody());
               if(luckyDog.isAutoAck()){
                   if(message.getDeliverMode()==2){
                       parent.getDiskDataCenter().deleteMessage(queue,message);
                   }
                   parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(),message.getMessageId());
                   parent.getMemoryDataCenter().removeMessage(message.getMessageId());
                   System.out.println("[ConsumerManager] 消息被成功消费！queueName="+queue.getName());
               }
           }catch (Exception e){
               e.printStackTrace();
           }
        });
    }
}
